package com.lx.db;

import com.lx.counter.CounterDriver;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: chenjiang
 * @Date: 2021/11/22/15:55
 * @Description:
 */
public class DBInputFormatReadDBDriver extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        DBConfiguration.configureDB(
                conf, "com.mysql.jdbc.Driver",
                "jdbc:mysql://127.0.0.1:3306/mr_shop",
                "root", "root"
        );
        Job job = Job.getInstance(conf, DBInputFormatReadDBDriver.class.getSimpleName());

        job.setJarByClass(DBInputFormatReadDBDriver.class);
        job.setMapperClass(ReadDBMapper.class);
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(Text.class);

        //设置InputFormat
        job.setInputFormatClass(DBInputFormat.class);


        final Path output = new Path(args[1]);

        DBInputFormat.setInput(job, GoodsEntity.class,
                "select goodsId,goodsSn,goodsName,marketPrice,shopPrice,saleNum from lanyxp_goods",
                "SELECT count(1) from lanyxp_goods");
        FileOutputFormat.setOutputPath(job, output);

        boolean result = job.waitForCompletion(true);
        return result ? 0 : 1;
    }

    class ReadDBMapper extends Mapper<LongWritable, GoodsEntity, LongWritable, Text> {
        Text text = new Text();

        @Override
        protected void map(LongWritable key, GoodsEntity value, Context context) throws IOException, InterruptedException {
            text.set(value.toString());
            context.write(key, text);
        }
    }
}
